#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include "logger.hpp"

namespace mag {

    class MQClient {
    public:
        // 定义消息回调函数类型：接收消息体和消息体大小
        using MessageCallback = std::function<void(const char*, size_t)>;
        // 定义智能指针类型，用于管理 MQClient 对象
        using ptr = std::shared_ptr<MQClient>;

        // 构造函数：初始化 AMQP 连接和事件循环
        MQClient(const std::string &user, 
                 const std::string &passwd, 
                 const std::string &host) {
            _loop = EV_DEFAULT;  // 使用默认事件循环
            _handler = std::make_unique<AMQP::LibEvHandler>(_loop);  // 初始化 LibEv 处理器

            // 构建 AMQP 连接地址，格式为 amqp://user:password@host:port/
            std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
            AMQP::Address address(url);
            _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);  // 创建连接
            _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());  // 创建信道

            // 启动事件循环的线程
            _loop_thread = std::thread([this]() {
                ev_run(_loop, 0);  // 启动事件循环
            });
        }

        // 析构函数：关闭事件循环并释放资源
        ~MQClient() {
            ev_async_init(&_async_watcher, watcher_callback);  // 初始化异步事件
            ev_async_start(_loop, &_async_watcher);  // 启动异步事件
            ev_async_send(_loop, &_async_watcher);  // 触发异步事件，停止事件循环
            _loop_thread.join();  // 等待事件循环线程结束
            _loop = nullptr;  // 清理事件循环指针
        }

        // 声明 AMQP 组件：交换机、队列、绑定等
        void declareComponents(const std::string &exchange,
                               const std::string &queue,
                               const std::string &routing_key = "routing_key",
                               AMQP::ExchangeType exchange_type = AMQP::ExchangeType::direct) {
            // 声明交换机
            _channel->declareExchange(exchange, exchange_type)
                .onError([](const char *message) {
                    LOG_ERROR("声明交换机失败：{}", message);
                    exit(0);  // 出错时退出
                })
                .onSuccess([exchange]() {
                    LOG_ERROR("{} 交换机创建成功！", exchange);
                });

            // 声明队列
            _channel->declareQueue(queue)
                .onError([](const char *message) {
                    LOG_ERROR("声明队列失败：{}", message);
                    exit(0);  // 出错时退出
                })
                .onSuccess([queue]() {
                    LOG_ERROR("{} 队列创建成功！", queue);
                });

            // 绑定交换机和队列
            _channel->bindQueue(exchange, queue, routing_key)
                .onError([exchange, queue](const char *message) {
                    LOG_ERROR("{} - {} 绑定失败：{}", exchange, queue, message);
                    exit(0);  // 出错时退出
                })
                .onSuccess([exchange, queue, routing_key]() {
                    LOG_ERROR("{} - {} - {} 绑定成功！", exchange, queue, routing_key);
                });
        }

        // 发布消息到指定交换机
        bool publish(const std::string &exchange, 
                     const std::string &msg, 
                     const std::string &routing_key = "routing_key") {
            LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange, routing_key);
            bool ret = _channel->publish(exchange, routing_key, msg);  // 发布消息
            if (ret == false) {
                LOG_ERROR("{} 发布消息失败：", exchange);
                return false;  // 如果发布失败，返回 false
            }
            return true;  // 发布成功，返回 true
        }

        // 消费队列中的消息，传入回调函数来处理接收到的消息
        void consume(const std::string &queue, const MessageCallback &cb) {
            LOG_DEBUG("开始订阅 {} 队列消息！", queue);
            _channel->consume(queue, "consume-tag")  // 返回值 DeferredConsumer
                .onReceived([this, cb](const AMQP::Message &message, 
                                       uint64_t deliveryTag, 
                                       bool redelivered) {
                    cb(message.body(), message.bodySize());  // 调用回调处理消息
                    _channel->ack(deliveryTag);  // 确认消息已处理
                })
                .onError([queue](const char *message) {
                    LOG_ERROR("订阅 {} 队列消息失败: {}", queue, message);
                    exit(0);  // 出错时退出
                });
        }

    private:
        // 异步事件回调函数，用于退出事件循环
        static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents) {
            ev_break(loop, EVBREAK_ALL);  // 退出事件循环
        }

    private:
        struct ev_async _async_watcher;  // 异步事件句柄
        struct ev_loop *_loop;  // 事件循环指针
        std::unique_ptr<AMQP::LibEvHandler> _handler;  // LibEv 事件处理器
        std::unique_ptr<AMQP::TcpConnection> _connection;  // AMQP 连接
        std::unique_ptr<AMQP::TcpChannel> _channel;  // AMQP 信道
        std::thread _loop_thread;  // 事件循环线程
    };

}
